package mr.stocks.question2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.stocks.avro.model.StocksValuesAvro;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroMultipleOutputs;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class StocksAvroUpdateReducer extends
        Reducer<AvroKey<StocksReportAvro>, NullWritable, AvroKey<StocksReportAvro>, NullWritable> {
    private StocksReportAvro resultReport;
    private AvroMultipleOutputs out;
    private StringBuilder sb = new StringBuilder();

    @Override
    public void setup(Context context) {
        out = new  AvroMultipleOutputs(context);
    }
    
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        out.close();
    }
    
    @Override
    protected void reduce(AvroKey<StocksReportAvro> key, Iterable<NullWritable> incoming, Context context)
            throws IOException, InterruptedException {

        resultReport = key.datum();
        sb.setLength(0);
        sb.append(resultReport.getYearMonthDay().getYear().toString() + "_");
        sb.append(resultReport.getYearMonthDay().getMonth().toString() + "_");
        sb.append(resultReport.getYearMonthDay().getDay().toString());
        out.write(key, NullWritable.get(), sb.toString());
    }
}
